package net.bwie.realtime.vehicle2.function;

import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.vehicle2.bean.VehicleData;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

/**
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class VehicleWindowBatteryHealthFunction implements WindowFunction<VehicleData, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<VehicleData> input, Collector<String> out) throws Exception {
        String window_start = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String window_end = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);
        ArrayList<String> list = new ArrayList<>();
        Double sum = 0.0;
        for (VehicleData vehicleData : input) {
            list.add(vehicleData.getVin());
            sum += vehicleData.getBatteryLevel();
        }
        int count = list.size();
        out.collect(window_start + "," + window_end + "," + (sum / count));
    }
}
